package cn.bossfriday.im.access.server.listener;

import cn.bossfriday.common.router.ClusterNode;
import cn.bossfriday.common.router.ClusterRouterFactory;
import cn.bossfriday.im.access.server.AccessContextAttributeKey;
import cn.bossfriday.im.access.server.MqttAccessCommon;
import cn.bossfriday.im.access.server.core.BaseMqttMessageListener;
import cn.bossfriday.im.common.codec.ImTokenCodec;
import cn.bossfriday.im.common.conf.ConfigurationAllLoader;
import cn.bossfriday.im.common.db.entity.AppInfo;
import cn.bossfriday.im.common.entity.ImToken;
import cn.bossfriday.im.common.entity.conf.GlobalConfig;
import cn.bossfriday.im.common.enums.access.ConnectState;
import cn.bossfriday.im.common.helper.AppHelper;
import cn.bossfriday.im.common.rpc.RpcMessageContext;
import cn.bossfriday.im.protocol.client.ClientInfo;
import cn.bossfriday.im.protocol.enums.ConnectionStatus;
import cn.bossfriday.im.protocol.message.ConnAckMessage;
import cn.bossfriday.im.protocol.message.ConnectMessage;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import static cn.bossfriday.im.access.server.AccessContextAttributeKey.*;
import static cn.bossfriday.im.common.constant.ApiConstant.API_ROUTE_KEY_USER_NAVI;

/**
 * ConnectMessageListener
 *
 * @author chenx
 */
@Slf4j
public class ConnectMessageListener extends BaseMqttMessageListener<ConnectMessage> {

    public ConnectMessageListener(ConnectMessage msg, ChannelHandlerContext ctx) {
        super(msg, ctx);
    }

    @Override
    public void onMqttMessageReceived() {
        try {
            boolean isChannelActive = this.ctx.channel().attr(IS_CHANNEL_ACTIVE).get();
            if (!isChannelActive) {
                this.ctx.close();
                return;
            }

            // 解析will
            ClientInfo clientInfo = ClientInfo.fromWill(this.msg.getWill());
            if (Objects.isNull(clientInfo) || Objects.isNull(clientInfo.getClientType())) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.IDENTIFIER_REJECTED, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            GlobalConfig globalConfig = ConfigurationAllLoader.getInstance().getGlobalConfig();

            // Token校验
            ImToken token = ImTokenCodec.decode(this.msg.getToken());
            if (Objects.isNull(token)
                    || StringUtils.isEmpty(token.getUserId())
                    || StringUtils.isEmpty(token.getDeviceId())) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.INVALID_TOKEN, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            // token与deviceId绑定检查
            String deviceId = clientInfo.getDeviceId();
            if (!token.getDeviceId().equalsIgnoreCase(deviceId)) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.DEVICE_ERROR, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            // token过期检查
            long tokenExpireTime = globalConfig.getTokenExpireTime();
            if ((System.currentTimeMillis() - token.getTime()) >= tokenExpireTime) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.TOKEN_EXPIRE, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            // app状态检查
            long appId = token.getAppId();
            boolean isAppOk = AppHelper.isAppOk(appId);
            if (!isAppOk) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.APP_BLOCK_OR_DELETE, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            // appSecret刷新所有老token失效；
            AppInfo appInfo = AppHelper.getAppInfo(appId);
            if (token.getAppSecretHash() != AppHelper.getAppSecretHashCode(appInfo.getAppSecret())) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.INVALID_TOKEN, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            // 接入服务落点检查
            String uid = token.getUserId();
            String clientIp = MqttAccessCommon.getClientIp(this.msg, this.ctx);
            String routableResourceId = RpcMessageContext.getRouteResourceId(appId, uid);
            ClusterNode clusterNode = ClusterRouterFactory.getClusterRouter().getTargetClusterNode(API_ROUTE_KEY_USER_NAVI, routableResourceId);
            if (!clusterNode.getName().equalsIgnoreCase(ConfigurationAllLoader.getInstance().getSystemConfig().getClusterNode().getName())) {
                MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.REDIRECT, ConnectState.CONN_ACK_FAILURE);
                return;
            }

            // 接受当前连接请求
            this.ctx.channel().attr(CONN_STATE).set(ConnectState.CONN);
            this.ctx.channel().attr(CLIENT_INFO).set(clientInfo);
            this.ctx.channel().attr(IS_CONNECTED).set(true);
            this.ctx.channel().attr(APPID).set(appId);
            this.ctx.channel().attr(USERID).set(uid);

            this.acceptConnection(token, clientIp);
        } catch (Exception ex) {
            log.error("ConnectMessageListener.onMqttMessageReceived() error!", ex);
            MqttAccessCommon.sendConnAck(this.ctx, ConnectionStatus.SERVICE_ERROR, ConnectState.CONN_ACK_FAILURE);
        }
    }

    /**
     * acceptConnection
     */
    private void acceptConnection(ImToken token, String clientIp) {
        final long time = System.currentTimeMillis();
        String clientInfo = "todo";
        ConnAckMessage connAck = new ConnAckMessage(ConnectionStatus.ACCEPTED.getCode());
        connAck.setUserId(token.getUserId());
        connAck.setSessionId(this.ctx.channel().attr(SESSION_ID).get());
        connAck.setTimestamp(time);

        MqttAccessCommon.writeAndFlush(connAck, this.ctx).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                this.ctx.channel().attr(AccessContextAttributeKey.CONN_STATE).set(ConnectState.CONN_ACK);
                this.ctx.channel().attr(AccessContextAttributeKey.CLIENT_MESSAGE_ID).set(0);
                this.ctx.channel().attr(AccessContextAttributeKey.SERVER_MESSAGE_ID).set(0);
                this.ctx.channel().attr(AccessContextAttributeKey.LAST_PUBLISH_COUNTER).set(0);
                this.ctx.channel().attr(AccessContextAttributeKey.LAST_PUBLISH_TIMESTAMP).set(time);
                this.ctx.channel().attr(AccessContextAttributeKey.PUBLISH_CALLBACK).set(new ConcurrentHashMap<>());
                this.ctx.channel().attr(AccessContextAttributeKey.QUERY_CALLBACK).set(new ConcurrentHashMap<>());

                log.info("[#UserConnectDone#]: appId: {}, uid: {},  deviceId: {}, clientIp: {}", token.getAppId(), token.getUserId(), token.getDeviceId(), clientIp);
            }
        });
    }
}
